package com.googlecode.quillen.service;

import com.googlecode.quillen.domain.*;
import static com.googlecode.quillen.util.Utils.logDebug;
import com.googlecode.quillen.util.Pair;
import com.googlecode.quillen.repository.AttributeStorage;
import com.googlecode.quillen.repository.ObjectStorage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.lang.StringUtils;

import java.security.NoSuchAlgorithmException;
import java.io.IOException;
import java.util.*;

/**
 * Created by IntelliJ IDEA.
 * User: greg
 * Date: Dec 8, 2008
 * Time: 7:31:23 PM
 */
public class ChunkServiceImpl implements ChunkService
{
    private static final Log LOG = LogFactory.getLog(ChunkServiceImpl.class);

    private final CompressionService compressionService;
    private final AttributeStorage attributeStorage;
    private final ObjectStorage objectStorage;

    public ChunkServiceImpl(CompressionService compressionService, AttributeStorage attributeStorage, ObjectStorage objectStorage)
    {
        this.compressionService = compressionService;
        this.attributeStorage = attributeStorage;
        this.objectStorage = objectStorage;
    }

    public void createStorage() throws AttributeStorageException, ObjectStorageException
    {
        attributeStorage.createDatabase(AttributeDatabase.chunks.toString());
        objectStorage.createBucket(ObjectBucket.chunks.toString());

        logDebug(LOG, "created chunk storage");
    }

    public void deleteStorage() throws AttributeStorageException, ObjectStorageException
    {
        attributeStorage.deleteDatabase(AttributeDatabase.chunks.toString());
        objectStorage.deleteBucket(ObjectBucket.chunks.toString());

        logDebug(LOG, "deleted chunk storage");
    }

    public Chunk get(String chunkKey) throws ObjectStorageException, NoSuchAlgorithmException, IOException
    {
        byte[] data = compressionService.decompress(objectStorage.get(ObjectBucket.chunks.toString(), chunkKey));

        if (data != null)
        {
            Chunk result = new Chunk(data);

            logDebug(LOG, "got chunk %s of size %d bytes", result.getKey(), result.getSize());

            return result;
        }
        else
        {
            return null;
        }
    }

    public void put(String shadowKey, Chunk chunk) throws ObjectStorageException, AttributeStorageException, IOException
    {
        // TODO: could use a cache (of chunkKey -> attributes) to potentially prevent this hit, make sure it's not unbounded

        Attributes attrs = coalesceBundles(chunk.getKey(), "size", shadowKey).first;

        Attributes newAttrs = new Attributes();

        if (attrs == null)
        {
            objectStorage.put(ObjectBucket.chunks.toString(), chunk.getKey(), compressionService.compress(chunk.getData()));
        }
        else if (!attrs.containsKey("size") || chunk.getSize() != Long.valueOf(attrs.getSingle("size")))
        {
            throw new IllegalStateException("existing size is missing or does not match for chunk " + chunk.getKey());
        }

        if (attrs == null || !Boolean.valueOf(attrs.getSingle(shadowKey)))
        {
            newAttrs.add("shadowKey", shadowKey);
            newAttrs.add(shadowKey, Boolean.TRUE.toString());

            putBundles(chunk, newAttrs);

            logDebug(LOG, "put chunk %s for shadow file %s", chunk.getKey(), shadowKey);
        }
    }

    public boolean delete(String chunkKey, Collection<String> shadowKeys) throws ObjectStorageException, AttributeStorageException
    {
        Pair<Attributes, Integer> pair = coalesceBundles(chunkKey);

        final Attributes attrs = pair.first;
        final int bundleCount = pair.second;

        if (attrs != null)
        {
            Set<String> shadowKeySet = new HashSet<String>(shadowKeys);
            Attributes toDelete = new Attributes();

            for (String shadowKey : new ArrayList<String>(attrs.get("shadowKey")))
            {
                if (shadowKeySet.isEmpty() || shadowKeySet.contains(shadowKey))
                {
                    toDelete.add("shadowKey", attrs.remove("shadowKey", shadowKey));
                }
            }

            if (attrs.get("shadowKey") != null)
            {
                for (int i = 0; i < bundleCount; i++)
                {
                    attributeStorage.delete(AttributeDatabase.chunks.toString(), chunkKey + "-" + i, toDelete);
                }

                logDebug(LOG, "removed chunk %s from shadow file %s", chunkKey, StringUtils.join(shadowKeys, ","));

                return false;
            }
            else
            {
                for (int i = 0; i < bundleCount; i++)
                {
                    attributeStorage.delete(AttributeDatabase.chunks.toString(), chunkKey + "-" + i);
                }

                objectStorage.delete(ObjectBucket.chunks.toString(), chunkKey);

                logDebug(LOG, "deleted chunk %s", chunkKey);

                return true;
            }
        }
        else
        {
            return false;
        }
    }

    private void putBundles(Chunk chunk, Attributes attributes) throws AttributeStorageException
    {
        for (int i = 0; !attributes.isEmpty(); i++)
        {
            attributes.add("size", Long.toString(chunk.getSize()));
            attributes.setPriority("size", 1);
            attributeStorage.put(AttributeDatabase.chunks.toString(), chunk.getKey() + "-" + i, attributes);
        }
    }

    private Pair<Attributes, Integer> coalesceBundles(String key, String... attributesToGet) throws AttributeStorageException
    {
        String[] attributesToGetArr = null;
        if (attributesToGet != null && attributesToGet.length > 0)
        {
            attributesToGetArr = new String[attributesToGet.length + 1];
            System.arraycopy(attributesToGet, 0, attributesToGetArr, 0, attributesToGet.length);
            attributesToGetArr[attributesToGet.length] = "size";
        }

        Attributes result = new Attributes();

        int i = 0;
        Attributes bundle;
        do
        {
            bundle = attributeStorage.get(AttributeDatabase.chunks.toString(), key + "-" + i++, attributesToGetArr);
            result.merge(bundle);
        }
        while (bundle != null);

        return Pair.get(result.isEmpty() ? null : result, i - 1);
    }
}
